#!/usr/bin/env python3
# coding=utf-8
"""
Python 3.14 多线程性能测试 - I/O密集型任务
测试文件读写操作的多线程性能
"""

import threading
import time
import os
import tempfile
import concurrent.futures
from typing import List


def io_intensive_task(file_path: str, data: str, iterations: int) -> int:
    """
    模拟I/O密集型任务 - 文件读写操作
    
    参数:
        file_path (str): 文件路径
        data (str): 写入数据
        iterations (int): 操作次数
    
    返回:
        int: 操作次数
    """
    total_bytes = 0
    for _ in range(iterations):
        # 写入文件
        with open(file_path, 'w', encoding='utf-8') as f:
            f.write(data)
        
        # 读取文件
        with open(file_path, 'r', encoding='utf-8') as f:
            content = f.read()
            total_bytes += len(content)
    
    return total_bytes


def single_thread_io_test(file_path: str, data: str, iterations: int) -> float:
    """单线程I/O测试"""
    start_time = time.time()
    result = io_intensive_task(file_path, data, iterations)
    end_time = time.time()
    print(f"单线程I/O结果: {result} 字节, 耗时: {end_time - start_time:.4f} 秒")
    return end_time - start_time


def multi_thread_io_test(num_threads: int, file_paths: List[str], data: str, iterations: int) -> float:
    """多线程I/O测试"""
    start_time = time.time()
    
    threads = []
    for i in range(num_threads):
        thread = threading.Thread(
            target=io_intensive_task, 
            args=(file_paths[i], data, iterations)
        )
        threads.append(thread)
        thread.start()
    
    for thread in threads:
        thread.join()
    
    end_time = time.time()
    print(f"{num_threads}线程I/O耗时: {end_time - start_time:.4f} 秒")
    return end_time - start_time


def thread_pool_io_test(num_threads: int, file_paths: List[str], data: str, iterations: int) -> float:
    """线程池I/O测试"""
    start_time = time.time()
    
    with concurrent.futures.ThreadPoolExecutor(max_workers=num_threads) as executor:
        futures = [
            executor.submit(io_intensive_task, file_paths[i], data, iterations)
            for i in range(num_threads)
        ]
        results = [future.result() for future in futures]
    
    end_time = time.time()
    print(f"线程池I/O({num_threads}线程)耗时: {end_time - start_time:.4f} 秒")
    return end_time - start_time


def io_performance_test():
    """I/O性能测试"""
    print("=== Python 3.14 I/O密集型多线程性能测试 ===")
    
    # 测试参数
    num_threads = 4
    iterations = 50
    data = "Python 3.14 多线程I/O性能测试数据。" * 1000
    
    # 创建临时文件
    temp_files = []
    for i in range(num_threads):
        temp_file = tempfile.NamedTemporaryFile(mode='w', delete=False, suffix=f'_test_{i}.txt')
        temp_file.close()
        temp_files.append(temp_file.name)
    
    try:
        # 单线程测试
        print("1. 单线程I/O测试:")
        single_time = single_thread_io_test(temp_files[0], data, iterations)
        
        print("\n2. 多线程I/O测试:")
        multi_time = multi_thread_io_test(num_threads, temp_files, data, iterations)
        
        print("\n3. 线程池I/O测试:")
        pool_time = thread_pool_io_test(num_threads, temp_files, data, iterations)
        
        # 性能分析
        print("\n=== I/O性能分析 ===")
        print(f"多线程I/O加速比: {single_time / multi_time:.2f}x")
        print(f"线程池I/O加速比: {single_time / pool_time:.2f}x")
        
        if multi_time < single_time:
            print("✅ I/O多线程性能提升明显！")
        else:
            print("⚠️  I/O多线程性能未提升")
    
    finally:
        # 清理临时文件
        for temp_file in temp_files:
            try:
                os.unlink(temp_file)
            except OSError:
                pass


if __name__ == "__main__":
    io_performance_test()